/*
 * Copyright 2018-2021 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.demo.common.redis.delay;

import cn.hutool.extra.spring.SpringUtil;

import com.demo.common.redis.key.RedisDataSpace;
import com.demo.common.redis.key.RedisKeyGenerator;
import com.demo.common.redis.key.RedisPublicKey;
import com.demo.common.redis.service.RedisService;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;

/**
 * 延迟消息监听基础实现。
 *
 * @author molong
 * @date 2022/6/22
 */
@Slf4j
public abstract class BaseDelayMessageListener<T> implements DelayMessageListener{

	/**
	 * 最大处理次数
	 */
	private static final long MAX_HANDLE_COUNT = 5;

	/**
	 * 序列化与反序列化对象
	 */
	private ObjectMapper objectMapper;

	/**
	 * 获取序列化与反序列化器
	 * @return ObjectMapper
	 */
	protected ObjectMapper getObjectMapper(){
		if(objectMapper == null){
			objectMapper = SpringUtil.getBean(ObjectMapper.class);
		}
		return objectMapper;
	}

	/**
	 * redis 操作对象
	 */
	private RedisService redisService;

	/**
	 * 获取 redis 操作对象
	 * @return  RedisTemplate<String, String>
	 */
	protected RedisService getRedisService(){
		if(redisService == null){
			redisService = SpringUtil.getBean(RedisService.class);
		}
		return redisService;
	}

	/**
	 * 获取反序列化具体的类型
	 * <code>
	 *     return new TypeReference<TestObject>() {};
	 * </code>
	 * @return 返回一个反序列化用的类型
	 */
	public abstract TypeReference<T> getTypeReference();

	@Override
	public void onMessage(DelayMessage message) {
		String payload = message.getPayload();
		try {
			log.info("---key：{}, 接收到延迟消息：消息当前处理次数： = {}", getRedisKey(), message.getHandleCount());
			//反序列化
			T t = getObjectMapper().readValue(payload, getTypeReference());
			handler(t);
		} catch (Exception e){
			log.error("消息处理异常：", e);
			//如果处理次数超过5次都没成功，那么此时将消息丢入死信
			if(message.getHandleCount() > MAX_HANDLE_COUNT){
				//丢入死信队列
				DelayDeadlyMessage delayDeadlyMessage = new DelayDeadlyMessage();
				delayDeadlyMessage.setCreateTime(LocalDateTime.now());
				delayDeadlyMessage.setHandleCount(message.getHandleCount());
				delayDeadlyMessage.setKey(getRedisKey());
				delayDeadlyMessage.setMsg(message);
				getRedisService().sAdd(RedisKeyGenerator.generateRedisKey(RedisDataSpace.PUBLIC, RedisPublicKey.DELAY_MSG_DEADLY), delayDeadlyMessage);
			}else {
				//处理失败，将消息再次存入队列中
				message.setHandleCount(message.getHandleCount() + 1);
				message.setDelayTime(LocalDateTime.now().plusSeconds(5));
				//延迟时间设置为当前时间后的5秒
				getRedisService().sendDelayMsg(getRedisKey(), message);
			}
		}
	}

	/**
	 * 消息处理
	 * @param content 消息内容
	 */
	public abstract void handler(T content);
}
